/**
* @Author  : zou.yingbin
* @Create  : 2022/1/10 10:56
* @Modify  : 2022/1/10 10:56
* @note    :
 */

package rabbitmq

import (
	"fmt"
	"github.com/streadway/amqp"
	"sync"
	"time"
)

//rabbitMQ 结构定义
type rabbitMQ struct {
	conn        *amqp.Connection // mq连接
	cfg         MqConfig         // 配置
	cl          []ReceiveChannel // 接收队列
	sender      SenderChannel    // 发送队列
	isConnected bool             // 是否已连接
	onMsg       MessageCallBack  // 消息处理
	logger      FieldLogger      // 日志组件指针
	bRun        bool             // 是否运行
	mtx         sync.Mutex
}

//connectMq 阻塞模式, 直到连接成功才返回
func (r *rabbitMQ) connectMq() {
	var err error
	cfg := r.cfg
	for true {
		r.logger.Info("try to connect rabbitmq:", cfg.Uri)
		r.conn, err = amqp.Dial(cfg.Uri)
		if err != nil {
			r.logger.Error(err)
			if r.bRun {
				r.logger.Info("\ntry connect rabbitmq after 1 seconds")
				time.Sleep(time.Second)
				continue
			}
		}
		break
	}
	r.setConnectStatus(true)
}

func (r *rabbitMQ) setConnectStatus(status bool) {
	r.mtx.Lock()
	r.isConnected = status
	r.mtx.Unlock()
}

//SubscribeTopic 订阅主题, 请在执行Start()前订阅
func (r *rabbitMQ) SubscribeTopic(topic string, autoAck, autoDelete bool) {
	c := ReceiveChannel{
		topic:      topic,
		autoAck:    autoAck,
		autoDelete: autoDelete,
		c:          nil,
		queueName:  fmt.Sprintf("%v%v_%v", r.cfg.ServerName, r.cfg.ServerID, topic),
	}
	r.cl = append(r.cl, c)
}

//Publish 发送消息
func (r *rabbitMQ) Publish(topic string, msg []byte) error {
	if !r.isConnected {
		err := fmt.Errorf("rabbitmq not connected")
		r.logger.Error(err)
		return err
	}
	return r.sender.c.Publish(r.cfg.ExchangeName, topic, false, false, amqp.Publishing{
		ContentType: "text/plain",
		Body:        msg,
	})
}

//SetLogger 设置日志组件
func (r *rabbitMQ) SetLogger(logger FieldLogger) {
	r.logger = logger
}

func (r *rabbitMQ) initReceiveChannel() error {
	for i := range r.cl {
		err := r.cl[i].init(r.conn, r.cfg, r.onMsg)
		if err != nil {
			return err
		}
	}
	return nil
}

func (r *rabbitMQ) handleErr() {
	c := make(chan *amqp.Error, 0)
	ec := r.conn.NotifyClose(c)
	err := <-ec
	r.logger.Error(err.Error())
	r.setConnectStatus(false)
	if r.bRun {
		go r.Start()
	}
}

//Start 启动总线
func (r *rabbitMQ) Start() {
	r.connectMq()
	var err error
	if err = r.initReceiveChannel(); err != nil {
		r.logger.Fatal(err)
	}
	if err = r.sender.init(r.conn, r.cfg); err != nil {
		r.logger.Fatal(err)
	}
}

//Stop 停止总线
func (r *rabbitMQ) Stop() {
	r.setConnectStatus(false)
	r.bRun = false
	for _, v := range r.cl {
		if v.autoDelete {
			v.deleteQueue()
		}
	}
	if r.conn != nil {
		_ = r.conn.Close()
	}
}
